import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.{Seconds, StreamingContext}

object Window {

  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setMaster("local[*]").setAppName(" ")
    val ssc = new StreamingContext(conf, Seconds(3))

    val windowDs: DStream[(String, Int)] = ssc.socketTextStream("usdp-o3tbdsfp-monitor1", 4444)
      .flatMap(_.split(" ")).map((_, 1))
      .window(Seconds(12), Seconds(6))


    windowDs
      .reduceByKey(_ + _)
      .print()


    windowDs
      .reduceByKeyAndWindow((a: Int, b: Int) => a + b, Seconds(12), Seconds(6))
      .print()


    ssc.start()
    ssc.awaitTermination()
  }
}
